package com.allen.flink.batch;

import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 功能: MapPartition 测试
 *
 * @date: 2020-03-24 12:54
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class MapPartition {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //产生数据
        DataSet<Long> ds = env.generateSequence(1, 20);
        
        ds.mapPartition(new MyMapPartitionFuction()).print();
    }
    
    public static class MyMapPartitionFuction implements MapPartitionFunction<Long, Long> {
        @Override
        public void mapPartition(final Iterable<Long> values, final Collector<Long> out) throws Exception {
            long count = 0;
            for (Long value : values) {
                count++;
            }
            out.collect(count);
        }
    }
}
